package com.example.mq.mqserver;

import com.example.mq.VirtualHost;
import com.example.mq.common.BinaryTool;
import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.common.net.*;
import com.example.mq.mqserver.core.BasicProperties;

import java.io.*;
import java.net.ConnectException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/*
 * TCP服务器
 * */
public class BrokerServer {
    private ServerSocket serverSocket;

    //当前brokerserver只有一个虚拟机
    private VirtualHost virtualHost = new VirtualHost("default");

    //保存当前与客户端的会话
    //<channelID,Socket>
    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();

    //引入线程池，处理多个客户的请求
    private ExecutorService executorService;

    //控制服务器是否运行
    private volatile boolean runnable = true;


    /*
     * 构造方法*/
    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port);
    }

    /*
     * 用来启动BrokerServer*/
    public void start() throws IOException {
        System.out.println("brokerServer starting !");
        executorService = Executors.newCachedThreadPool();

        try{
            while (runnable) {//只要brokerserver启动，就一直监听，一旦有tcp连接，放入线程池执行相应逻辑

                // 你是把这个关了。这就是我一开始说的，accept 的时候，你把socket手动关了
                Socket clientSocket = serverSocket.accept();

                //把处理连接请求的逻辑丢给线程池
                executorService.submit(() -> {

                    processConnection(clientSocket);

                });
            }
        }catch (SocketException e){
//            e.printStackTrace();
            System.out.println("[BrokerServer] 服务器停止运行!");
        }catch (RuntimeException e){
            System.out.println("processConnection出现异常");
        }

    }

    //停止brokerserver
    public void stop() throws IOException {
        runnable = false;
        executorService.shutdownNow();//停止线程池now是强制关闭 shutdown是优雅关闭
        serverSocket.close();
    }

    //用来处理一个客户端的连接
    //连接中可能会涉及多个请求和响应
    private void processConnection(Socket clientSocket) {
        try (InputStream inputStream = clientSocket.getInputStream();
             OutputStream outputStream = clientSocket.getOutputStream()) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream);
                 DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                while (true) {

                    //1、读取请求，并且解析
                    Request reqeust = readRequest(dataInputStream);

                    //2、根据请求，计算响应
                    Response response = process(reqeust, clientSocket);

                    //3、响应写会给客户端
                    writeResponse(dataOutputStream, response);
                }
            } catch (EOFException | SocketException e) {
                //获取这个异常，说明已经文件读取完毕(借助异常，停止while循环)
                System.out.println("BrokerServer连接关闭 ！客户端地址：" + clientSocket.getInetAddress().toString() + " 端口号：" + clientSocket.getPort());
            } catch (ClassNotFoundException|IOException|MqException e) {
                System.out.println("[BrokerServer] connection 出现异常!");
                throw new RuntimeException(e);
            }


        } catch (IOException e) {
            System.out.println("[BrokerServer.processConnection] 服务器连接出现问题");
            throw new RuntimeException(e);
        } finally {
            //连接处理完毕，关闭连接(Connection)
            try {
                // 应该是关闭当前的socket连接，而不是关闭serverSocket
                // serverSocket 一关，后面所有的连接，都连不上了
                clientSocket.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }

            //清理掉channel
            clearClosedSession(clientSocket);
        }
    }


    //    响应写回给客户端(把response放到流中)
    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {

        //写入类型、长度、载荷
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());

        //刷新缓冲区非常重要
        dataOutputStream.flush();
    }


    //读取请求，并且解析(根据我们自定义的应用层协议)
    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();
        request.setType(dataInputStream.readInt());
        request.setLength(dataInputStream.readInt());
        byte[] payload = new byte[request.getLength()];
        int actualLength = dataInputStream.read(payload);
        if (actualLength != request.getLength()) {
            throw new IOException("读取请求格式出错");
        }
        request.setPayload(payload);
        return request;
    }


    //根据请求，计算响应z(request解析为指定Java对象，然后写入Socket发送给客户端)
    private Response process(Request reqeust, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {

        //1、把request解析为参数父类basicArguments
        BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(reqeust.getPayload());
        boolean ok = true;

        //2、根据不同的type，把消息写入socket
        if (reqeust.getType() == 0x1) {//创建channel

            sessions.put(basicArguments.getChannelId(), clientSocket);
            System.out.println("[BrokerServer] 创建 channel 完成! channelId=" + basicArguments.getChannelId());

        } else if (reqeust.getType() == 0x2) {//销毁channel

            sessions.remove(basicArguments.getChannelId());
            System.out.println("[BrokerServer] 销毁 channel 完成! channelId=" + basicArguments.getChannelId());

        } else if (reqeust.getType() == 0x3) {//创建交换机
            //payLoad就是exchangeDeclareArgument对象



            ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) BinaryTool.fromBytes(reqeust.getPayload());
            System.out.println("开始声明交换机 没有进入virtualHost.exchangeDeclare前，exchangeName="+arguments.getExchangeName());

            ok = virtualHost.exchangeDeclare(arguments.getExchangeName(), arguments.getExchangeType(),
                    arguments.isDurable(), arguments.isAutoDelete(), arguments.getArguments());

        } else if (reqeust.getType() == 0x4) {//销毁交换机

            ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) BinaryTool.fromBytes(reqeust.getPayload());
            ok = virtualHost.exchangeDelete(arguments.getExchangeName());

        } else if (reqeust.getType() == 0x5) {//创建队列


            QueueDeclareArguments arguments = (QueueDeclareArguments) BinaryTool.fromBytes(reqeust.getPayload());
            ok = virtualHost.queueDeclare(arguments.getQueueName()
                    , arguments.isDurable()
                    , arguments.isExclusive()
                    , arguments.isAutoDelete()
                    , arguments.getArguments());

        } else if (reqeust.getType() == 0x6) {//删除队列

            QueueDeleteArguments arguments = (QueueDeleteArguments) BinaryTool.fromBytes(reqeust.getPayload());
            ok = virtualHost.queueDelete(arguments.getQueueName());

        } else if (reqeust.getType() == 0x7) {//创建绑定

            QueueBindArguments arguments = (QueueBindArguments) BinaryTool.fromBytes(reqeust.getPayload());
            ok = virtualHost.queueBind(arguments.getQueueName(), arguments.getExchangeName(), arguments.getBindingKey());

        } else if (reqeust.getType() == 0x8) {//删除绑定

            QueueUnbindArguments arguments = (QueueUnbindArguments) BinaryTool.fromBytes(reqeust.getPayload());
            ok = virtualHost.queueUnbind(arguments.getQueueName(), arguments.getExchangeName());

        } else if (reqeust.getType() == 0x9) {//发布消息

            BasicPublishArguments arguments = (BasicPublishArguments) BinaryTool.fromBytes(reqeust.getPayload());

            ok = virtualHost.basicPublish(arguments.getExchangeName(), arguments.getRoutingKey(), arguments.getProperties(), arguments.getBody());
        } else if (reqeust.getType() == 0xA) {//消费订阅


            BasicConsumeArguments arguments = (BasicConsumeArguments) BinaryTool.fromBytes(reqeust.getPayload());

            System.out.println("进入到消费消息的逻辑arguments+  "+arguments);
            System.out.println("进入到消费消息的逻辑getConsumerTag+  "+arguments.getConsumerTag());

            System.out.println("进入到消费消息的逻辑+getQueueName  "+arguments.getQueueName());
            System.out.println("进入到消费消息的逻辑+getChannelId  "+arguments.getChannelId());
            System.out.println("---------------------------------------------------------------------------------------");
            ok = virtualHost.basicConsume(
                    arguments.getConsumerTag()
                    , arguments.getQueueName()
                    , arguments.isAutoAck(),
                    new Consumer() {

                        //回调函数：要把服务器收到的消息，推送给消费者客户端
                        @Override//这里执行了回调函数
                        public void handleDelivery(String consumerTag, BasicProperties properties, byte[] body) throws MqException, IOException {
                            //需要知道收到的这个消息需要给那个客户端
                            //实际上channelId就是consumerTag，根据channelId查找指定socket即可
                            System.out.println("------------------session: "+sessions);
                            System.out.println("------------------consumerTag: "+consumerTag);
                            System.out.println("------------------body: "+new String(body));

                            //1)根据channelId找到Socket对象
                            Socket socket = sessions.get(consumerTag);
                            if (socket == null || socket.isClosed()) {//不存在或者已经关闭，出现了问题
                                throw new MqException("[BrokerServer] 订阅消费者的客户端连接（socket）已经关闭，无法发送消息给消费者");
                            }

                            //2)构造响应数据
                            SubScribeReturns returns = new SubScribeReturns();
                            returns.setConsumerTag(consumerTag);
                            returns.setBody(body);



                            returns.setBasicProperties(properties);
                            returns.setChannelId(consumerTag);
                            returns.setRid("");//这里只有请求，没有响应，不需要对应，按时不需要Rid
                            returns.setOk(true);
                            byte[] payload = BinaryTool.toBytes(returns);

                            Response response = new Response();
                            response.setType(0xC);//服务器给客户端推送实际MQ的消息
                            response.setLength(payload.length);

                            //这里的payload就是subscribeReturn
                            response.setPayload(payload);

                            //数据写回socket返回给客户端
                            /*注意！ dataOutputStream不能close
                             * 原因：他的关闭会触发clientSocket的关闭，导致后续无法持续写入消息给消费者*/
                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                            //0xc
                            writeResponse(dataOutputStream, response);

                        }
                    });

        } else if (reqeust.getType() == 0xB) {//调用basicAck确认消息

            BasicAckArguments arguments = (BasicAckArguments) BinaryTool.fromBytes(reqeust.getPayload());
            ok = virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());

        } else {//其他type类型，抛出异常

            throw new MqException("[BrokerServer] 位置的type type=" + reqeust.getType());

        }


        //3、构造返回对象的内容
        BasicReturns returns = new BasicReturns();
        returns.setChannelId(basicArguments.getChannelId());
        returns.setRid(basicArguments.getRid());//确保响应 服务一致是同一个
        returns.setOk(ok);//告诉客户端这次远程调用是否成功

        //3、构造Response（returns是统一结果返回）
        Response response = new Response();
        byte[] payload = BinaryTool.toBytes(returns);
        response.setType(reqeust.getType());
        response.setLength(payload.length);
        response.setPayload(payload);

        System.out.println("[Response] rid=" + returns.getRid() + ", channelId=" + returns.getChannelId()
                + ", type=" + response.getType() + ", length=" + response.getLength());
        return response;
    }

    //把channel清理掉（socket关闭后，清理sessions）
    private void clearClosedSession(Socket clientSocket) {

        //先枚举完，在统一删除，避免迭代出现问题
        List<String> toDeleteChannelId = new ArrayList<String>();

        for (Map.Entry<String, Socket> entry : sessions.entrySet()) {
            if (entry.getValue() == clientSocket) {
                toDeleteChannelId.add(entry.getKey());
            }
        }
        for (String channelId : toDeleteChannelId) {
            sessions.remove(channelId);
        }
        System.out.println("[BrokerServer] 清理 session 完成! 被清理的 channelId=" + toDeleteChannelId);

    }

}
